package cn.dengta.webapp.risk.service;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import javax.annotation.Nullable;

import cn.dengta.common.context.Env;
import cn.dengta.common.model.Messager;
import cn.dengta.common.model.Status;
import cn.dengta.common.util.SpringUtil;
import cn.dengta.context.model.Member;
import cn.dengta.context.model.WesKit;
import cn.dengta.context.web.ProxyRequest;
import cn.dengta.context.web.RequestContextImpl;
import cn.dengta.webapp.risk.bean.*;
import cn.dengta.webapp.risk.config.RiskConfiguration;
import cn.dengta.webapp.risk.entity.RiskConfig;
import cn.dengta.webapp.risk.util.SmHttpUtil;
import cn.dengta.webapp.risk.vo.SmRiskVO;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import me.codeplayer.util.*;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.core.task.TaskExecutor;

@Slf4j
public class RiskService {

	static TaskExecutor taskExecutor;
	static RiskConfigService configService;

	/**
	 * 只有接口调用成功时，才会返回 status == "OK" 以及对应的数据实体
	 */
	@Nullable
	public static Messager<SmRiskVO> runRisk(WesKit wesKit, @Nullable String userId, RiskType type, String value, RiskEvent event, @Nullable String toUserId) {
		final RiskRequestParam param = new RiskRequestParam(userId, type, event).setToUserId(toUserId);
		return doRunRisk(wesKit, param, value);
	}

	@Nullable
	public static Messager<SmRiskVO> runRisk(WesKit wesKit, @Nullable Long userId, RiskType type, String value, RiskEvent event) {
		return runRisk(wesKit, Member.encode(userId), type, value, event, null);
	}

	/**
	 * 数美天网风控检测
	 * 如果存在风险则返回 true
	 */
	public static boolean runSkynet(WesKit kit, RiskRequestParam param) {
		if (configService == null) {
			configService = SpringUtil.getBean(RiskConfigService.class);
		}
		// 获取风控开关，没有开启风控则直接忽略
		final RiskType riskType = RiskType.SKYNET;
		RiskConfig config = configService.get(kit, riskType);
		if (config == null || !config.isEnabled()) {
			return false;
		}
		param.setType(riskType);
		Messager<SmRiskVO> msger = doRunRisk(kit, param, param.getNickname());
		return SmRiskVO.hasRisk(msger, config.getPassLevel());
	}

	/**
	 * 只有接口调用成功时，才会返回 status == "OK" 以及对应的数据实体
	 *
	 * @param value 文字内容/图片链接/音视频文件链接
	 */
	@Nullable
	public static Messager<SmRiskVO> doRunRisk(WesKit wesKit, RiskRequestParam param, String value) {
		prepareContextParam(param);

		RiskBase requestEntity = createRequestEntity(param, value);
		if (requestEntity != null) {
			// TODO 可考虑在后台风控配置中增加风控类型对应的接口地址、风险类型，而不是写在枚举中
			return SmHttpUtil.post(wesKit, param.getType(), JSON.toJSONString(requestEntity));
		}
		return null;
	}

	/**
	 * 并行执行风控检测
	 *
	 * @see #doRunRisk(WesKit, RiskRequestParam, String)
	 */
	@Nullable
	public static List<Messager<SmRiskVO>> parallelRisk(WesKit kit, RiskRequestParam param, String[] values) {
		if (!X.isValid(values)) {
			return null;
		}

		if (values.length == 1) {
			final Messager<SmRiskVO> result = doRunRisk(kit, param, values[0]);
			return Collections.singletonList(result);
		}

		// 要先准备上下文参数，否则在异步环境中可能无法获取
		prepareContextParam(param);
		return multiParallel(values, val -> createFuture(() -> doRunRisk(kit, param, val)));
	}

	/**
	 * @param value 文字内容/图片链接/音视频文件链接
	 */
	protected static RiskBase createRequestEntity(RiskRequestParam param, final String value) {
		RiskBase riskBase = null;
		final AbstractRiskData data;
		final RiskEvent event = param.getEvent();
		switch (param.getType()) {
			case TEXT:
			case IM_TEXT:
			case IM_GROUP_TEXT:
				RiskText text = new RiskText();
				data = text.getData();
				if (event != null) {
					text.setEventId(event.value);
				}
				riskBase = text;
				break;
			case IMAGE:
			case MULTI_IMAGE:
			case IM_IMAGE:
			case IM_GROUP_IMAGE:
				RiskImage image = new RiskImage();
				data = image.getData();
				if (event != null) {
					image.setEventId(event.value);
				}
				riskBase = image;
				break;
			case VIDEO:
				RiskVideo video = new RiskVideo();
				data = video.getData();

				String btId = param.getRequestId();
				if (btId == null) {
					btId = Encrypter.md5(value);
				}

				video.setBtId(btId);
				video.setCallback(param.getCallback());
				video.getCallbackParam().setVerifyId(param.getVerifyId());
				riskBase = video;
				break;
			case SKYNET:
				RiskSkynet skynet = new RiskSkynet();
				skynet.setEventId(event.value);
				RiskSkynet.SkynetData skynetData = skynet.getData();
				final String appId = param.getAppId();
				if (StringUtil.notEmpty(appId)) {
					final String os = StringUtils.substringBetween(appId, "/", "/").toLowerCase();
					skynetData.setOs("wap".equals(os) ? "web" : os);
					skynetData.setAppVersion(ProxyRequest.getVersion(appId));
				}
				String phone = param.getPhone();
				if (StringUtil.notEmpty(phone)) {
					final String[] parts = StringUtils.split(phone, '-');
					phone = parts[parts.length - 1];
					String countryCode = parts.length == 2 ? parts[0] : "0086";
					skynetData.setCountryCode(countryCode);
					skynetData.setPhoneSha256(DigestUtils.sha256Hex(phone));
					skynetData.setPhoneMd5(DigestUtils.md5Hex(phone));
					skynetData.setPhone(phone);
				}
				SkynetType skynetType = X.expectNotNull(param.getSkynetType(), SkynetType.phoneMessage);
				skynetData.setType(skynetType.name());

				switch (event) {
					case REGISTER:
						skynetData.setNickName(param.getNickname())
								.setIsPhoneExist(Status.NO.value);
						if (StringUtils.isNotEmpty(param.getEmail())) {
							skynetData.setEmail(param.getEmail());
						}
						break;
					case LOGIN:
						// skyNetData.setLevel(0);
						skynetData.setValid(Status.YES.value);
						break;
				}
				data = skynetData;
				riskBase = skynet;
				break;
			default:
				data = null;
		}
		if (data != null) {
			data.init(param.getUserId(), value, StringUtil.isEmpty(param.getIp()) ? null : param.getIp(), param.getDeviceId());
			if (event == RiskEvent.MESSAGE && param.getToUserId() != null) {
				final JSONObject extra = data.extra();
				extra.put("receiveTokenId", param.getToUserId());
			}
		}
		return riskBase;
	}

	static void prepareContextParam(RiskRequestParam param) {
		if (param.getUserId() == null || param.getIp() == null) {
			RequestContextImpl req = RequestContextImpl.get();

			if (param.getUserId() == null) {
				final Member sessionUser = req.sessionUser();
				String userId = sessionUser != null ? sessionUser.encodedId() : UUID.randomUUID().toString();
				param.setUserId(userId);
			}

			if (param.getIp() == null) {
				String ip = !Env.inLocal() && !Env.inDev() // IP字段先排除本地环境，本地环境加上IP会调不通数美接口
						? req.clientIP()
						: ""; // 避免重复被初始化为 null

				param.setIp(ip);
			}
		}
	}

	/**
	 * 多个并行调用风控接口
	 */
	@SuppressWarnings("unchecked")
	private static List<Messager<SmRiskVO>> multiParallel(String[] values, Function<String, CompletableFuture<Messager<SmRiskVO>>> function) {
		CompletableFuture<Messager<SmRiskVO>>[] futures = Arrays.stream(values).map(function).toArray(CompletableFuture[]::new);
		CompletableFuture<Void> allFuture = CompletableFuture.allOf(futures);
		CompletableFuture<List<Messager<SmRiskVO>>> allFutureList = allFuture.thenApply(t -> Arrays.stream(futures).map(future -> {
			try {
				return future.get();
			} catch (Exception e) {
				log.error("调用风控接口出现异常", e);
				return null;
			}
		}).collect(Collectors.toList()));
		try {
			return allFutureList.get();
		} catch (Exception e) {
			log.error("获取风控结果出现异常", e);
		}
		return null;
	}

	private static CompletableFuture<Messager<SmRiskVO>> createFuture(Supplier<Messager<SmRiskVO>> supplier) {
		if (taskExecutor == null) {
			taskExecutor = SpringUtil.getBean(RiskConfiguration.RISK_TASK_EXECUTOR_BEAN_NAME, TaskExecutor.class);
		}
		return CompletableFuture.supplyAsync(supplier, taskExecutor);
	}

	/**
	 * 调用数美天网风险监测
	 * 如果有风险，则返回 true
	 */
	public static boolean skynetCheck(Long userId, String phone, String nickname, @Nullable String email, final boolean loginOrRegister,
	                                  String appId, String deviceId, final SkynetType inputSkynetType, final String ip, String remark, WesKit kit) {
		String encodedUserId = Member.encode(userId);
		boolean isAutoLogin = false;
		SkynetType skynetType = inputSkynetType;
		if (skynetType == SkynetType.fastlogin) {
			isAutoLogin = true;
			// 数美不支持 fastlogin 参数，此处仅用于区分是否为自动登录触发的天网验证
			skynetType = SkynetType.phoneMessage;
		}
		RiskEvent riskEvent = loginOrRegister ? RiskEvent.LOGIN : RiskEvent.REGISTER;
		RiskRequestParam requestParam = new RiskRequestParam(encodedUserId, riskEvent, appId, phone, nickname, deviceId)
				.setEmail(email);
		requestParam.setSkynetType(skynetType);
		if (StringUtil.notEmpty(ip)) {
			requestParam.setIp(ip);
		}
		boolean hasRisk = runSkynet(kit, requestParam);
		log.info("【天网】风控：{}风险，type={}，用户ID={}，deviceId={}，appId={}，phone={}，email={}，IP={}，remark={}",
				hasRisk ? "有" : "无", inputSkynetType, userId, deviceId, appId, phone, email, ip, StringUtil.toString(remark));
		if (isAutoLogin && hasRisk) {
			// 自动登录时不直接抛异常，需要在调用端进行强制下线处理
			return true;
		}
		// 系统识别到异常，阻断本次操作
		Assert.notTrue(hasRisk, "The system recognizes a RISK and interrupts the operation!");
		return false;
	}

	/**
	 * 音频，视频通话数美风控请求
	 */
	public static Messager<SmRiskVO> riskStream(WesKit wesKit, RiskType riskType, RiskBase dto) {
		return SmHttpUtil.post(wesKit, riskType, JSON.toJSONString(dto));
	}

	/**
	 * 音频，视频通话关闭通知
	 */
	public static Messager<SmRiskVO> closerRiskStream(WesKit wesKit, RiskType riskType, Map<String, String> closerMap) {
		return SmHttpUtil.post(wesKit, riskType, JSON.toJSONString(closerMap));
	}

}
